package com.linkstec.mongo4;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.bson.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

//import com.linkstec.kafka.KafkaSender;
import com.linkstec.util.ConfigUtil;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.MongoNamespace;
//import com.mongodb.MongoClientURI;
import com.mongodb.ReadPreference;
import com.mongodb.ServerAddress;
import com.mongodb.client.ChangeStreamIterable;
import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.changestream.ChangeStreamDocument;
import com.mongodb.client.model.changestream.FullDocument;
import com.mongodb.client.model.changestream.OperationType;

public class MongoDbUtil4 {

	private static final Logger logger = LoggerFactory.getLogger(MongoDbUtil4.class);
	private static String hostsAndPorts;
	private static MongoClient mongoClient = null;
	private static MongoDatabase testDatabase;
	private static MongoCollection<Document> collection;

	static {
//		hostsAndPorts = ConfigUtil.getProperty("mongo.config.replica.set.address", "localhost:27017");
		hostsAndPorts = ConfigUtil.getProperty("mongo4", "localhost:27017,localhost:27018,localhost:27019");
//		hostsAndPorts = "localhost:27017";
//		hostsAndPorts = "192.168.10.141:30000,192.168.10.143:30000,192.168.10.61:30000";
	}
	// connect to mongodb
	public static MongoClient getMongoClient() {
		if (null == mongoClient) {

			MongoClientOptions.Builder build = new MongoClientOptions.Builder();
			build.connectionsPerHost(50); // 与目标数据库能够建立的最大connection数量为50
//			// 如果当前所有的connection都在使用中，则每个connection上可以有50个线程排队等待
//			build.threadsAllowedToBlockForConnectionMultiplier(50);
			/*
			 * 一个线程访问数据库的时候，在成功获取到一个可用数据库连接之前的最长等待时间为2分钟
			 * 这里比较危险，如果超过maxWaitTime都没有获取到这个连接的话，该线程就会抛出Exception
			 * 故这里设置的maxWaitTime应该足够大，以免由于排队线程过多造成的数据库访问失败
			 */
			build.maxWaitTime(1000 * 60 * 2);
			build.connectTimeout(1000 * 60 * 1); // 与数据库建立连接的timeout设置为1分钟
			//读写分离
			build.readPreference(ReadPreference.secondary());
//			build.socketKeepAlive(true);//最关键，保证长时间不用不出错

			MongoClientOptions myOptions = build.build();

			// mongodb集群
			List<ServerAddress> addresses = new ArrayList<ServerAddress>();
//			hostsAndPorts = ConfigUtil.getProperty("mongo.hostsandports", "10.189.48.3:27017");
			for (String hostAndPort :hostsAndPorts.split(",")){
				String host = hostAndPort.split(":")[0];
				int port = Integer.valueOf(hostAndPort.split(":")[1]);
				ServerAddress address = new ServerAddress(host, port);
				addresses.add(address);
			}
			//复制集
//			MongoClientURI connectionString = new MongoClientURI("mongodb://localhost:27017,localhost:27018,localhost:27019/admin?replicaSet=shared1");

			mongoClient = new MongoClient(addresses, myOptions);
//			mongoClient = new MongoClient(connectionString);
			return mongoClient;
		} else {
			return mongoClient;
		}
	}

	public static MongoDatabase getMongoDatabase(String mongoDatabase) {
		mongoClient = getMongoClient();
		testDatabase = mongoClient.getDatabase(mongoDatabase);
		logger.info("获得testDatabase");
		return testDatabase;
	}

	public static MongoCollection<Document> getMongoCollection(MongoDatabase mongoDatabase, String collectionName) {
		collection = mongoDatabase.getCollection(collectionName);
		return collection;
	}
	public static MongoCollection<Document> getMongoCollection(String mongoDatabase, String collectionName) {
		mongoClient = getMongoClient();
		testDatabase = mongoClient.getDatabase(mongoDatabase);
		collection = testDatabase.getCollection(collectionName);
		return collection;
	}
	public static void main(String[] args) {
		getMongoClient();
		logger.info("开始");

		/*ChangeStreamIterable<Document> cSi = mongoClient.watch().fullDocument(FullDocument.UPDATE_LOOKUP);
		MongoCursor<ChangeStreamDocument<Document>> mongoCursor = cSi.iterator();
		while (mongoCursor.hasNext()) {
			logger.info("检测到变更：" + cSi);
			ChangeStreamDocument<Document> EventRepeat = mongoCursor.next();
			logger.info(EventRepeat + "");
		}*/
//		mongoClient.watch().fullDocument(FullDocument.UPDATE_LOOKUP).forEach(
//		(Consumer<ChangeStreamDocument<?>>)doc ->logger.info(doc + ""));

		//上面db级别，下面集合级别
		MongoCollection<Document> users = getMongoCollection("MOT","hello");
		ChangeStreamIterable<Document> cSi = users.watch().fullDocument(FullDocument.UPDATE_LOOKUP);
		logger.info("watch：" + cSi);
		MongoCursor<ChangeStreamDocument<Document>> mongoCursor = cSi.iterator();
		logger.info("iterator");
		while (mongoCursor.hasNext()) {
			logger.info("检测到变更：");
			ChangeStreamDocument<Document> EventRepeat = mongoCursor.next();
			//delete为null 更新operationType=OperationType{value='replace'}
			Document fullDocument = EventRepeat.getFullDocument();
			OperationType operationType = EventRepeat.getOperationType();
			MongoNamespace mongoNamespace = EventRepeat.getNamespace();

			String topic = mongoNamespace.getCollectionName();
			String operationTypeStr ="";
			operationTypeStr = operationType.getValue();
			if (null == fullDocument) {
				fullDocument = new Document();
			}
			fullDocument.put("operationType", operationTypeStr);
			logger.info(EventRepeat + "");
			logger.info(fullDocument + "");
			logger.info(operationTypeStr);
			logger.info(topic);
			/*try {
				KafkaSender.sendKafkaMessage(topic, fullDocument);
			} catch (Exception e) {
				e.printStackTrace();
			}*/
		}
		/*users.watch().fullDocument(FullDocument.UPDATE_LOOKUP).forEach(
		(Consumer<ChangeStreamDocument<?>>)doc ->logger.info(doc + ""));

		ChangeStreamIterable<Document> cSi2 = users.watch().fullDocument(FullDocument.UPDATE_LOOKUP);
		cSi2.forEach(
		(Consumer<ChangeStreamDocument<?>>)doc ->logger.info(doc + ""));*/

		// 取出查看
		MongoCollection<Document> users1 = getMongoCollection("MOT","hello");
		List<String> list = new ArrayList<String>();
		String sss = "王五,李四";
//		list.add("王五");
//		list.add("李四");
		list = Arrays.asList(sss.split(","));
        FindIterable<Document> iterable = users1.find(Filters.and(
				Filters.in("name", list),
				Filters.eq("add", "上海")));
        MongoCursor<Document> mongoCursor1 = iterable.iterator();
        /*while (iterable.iterator().hasNext()){
        	Document EventRepeat = iterable.iterator().next();
        	logger.info(EventRepeat.toJson());
        }*/
        while (mongoCursor1.hasNext()){
        	Document EventRepeat = mongoCursor1.next();
        	logger.info(EventRepeat.toJson());
        }
	}
}